package com.zengjianjun.storm.demo;

import com.zengjianjun.storm.demo.config.StormConfig;
import com.zengjianjun.storm.demo.wordcount.CountBolt;
import com.zengjianjun.storm.demo.wordcount.DataSourceSpout;
import com.zengjianjun.storm.demo.wordcount.SplitBolt;
import lombok.extern.slf4j.Slf4j;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.topology.TopologyBuilder;

/**
 * @author zengjianjun
 */
@Slf4j
public class WordCountStorm {

    public static void localStart() {
        try {
            log.info("======================[本地集群]词频统计启动开始======================");
            TopologyBuilder builder = new TopologyBuilder();
            builder.setSpout("DataSourceSpout", new DataSourceSpout());
            // 指明将 DataSourceSpout 的数据发送到 SplitBolt 中处理
            builder.setBolt("SplitBolt", new SplitBolt()).shuffleGrouping("DataSourceSpout");
            //  指明将 SplitBolt 的数据发送到 CountBolt 中 处理
            builder.setBolt("CountBolt", new CountBolt()).shuffleGrouping("SplitBolt");
            // 创建本地集群用于测试 这种模式不需要本机安装 storm
            //使用LocalCluster集群提交拓扑
            Config config = StormConfig.localConfig();
            LocalCluster cluster = new LocalCluster();
            cluster.submitTopology("word-count", config, builder.createTopology());
            log.info("======================[本地集群]词频统计启动结束======================");
        } catch (Exception e) {
            log.error("StormClient.wordCountSubmitTopology error:", e);
        }
    }

    public static void remoteStart() {
        try {
            log.info("======================[服务器集群]词频统计启动开始======================");
            TopologyBuilder builder = new TopologyBuilder();
            builder.setSpout("DataSourceSpout", new DataSourceSpout());
            // 指明将 DataSourceSpout 的数据发送到 SplitBolt 中处理
            builder.setBolt("SplitBolt", new SplitBolt()).shuffleGrouping("DataSourceSpout");
            //  指明将 SplitBolt 的数据发送到 CountBolt 中 处理
            builder.setBolt("CountBolt", new CountBolt()).shuffleGrouping("SplitBolt");
            Config config = StormConfig.remoteConfig();
            // 使用 StormSubmitter 提交 Topology 到服务器集群
            StormSubmitter.submitTopology("word-count", config, builder.createTopology());
            log.info("======================[服务器集群]词频统计启动结束======================");
        } catch (Exception e) {
            log.error("StormClient.wordCountSubmitTopology error:", e);
        }
    }

    public static void main(String[] args) {
        remoteStart();
    }
}
